package com.tsf.hbase;

import com.tsf.hbase.util.HBaseUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;

/**
 * 基于单行操作的唯一键。
 * <p>
 * 它是由{@link #unique}_{@link #time}_{@link #args}组成。
 *
 * @author taoningbo
 * @since 1.0.0
 */
public final class RowKey {

    /**
     * 索引字符，一般用于检索的关键词
     */
    private final String unique;

    /**
     * 时间，用于按照时间查询时使用，当有值时必须大于零，等于0时表示索引中无此参数
     */
    private Long time;

    /**
     * 可选参数，这些参数会作为时间之后的关键词
     */
    private List<String> args = null;

    /**
     * 是否对索引字符{@link #unique}进行倒序
     */
    private boolean reversed = false;

    /**
     * 是否对索引字符{@link #unique}采用Hash算法
     */
    private boolean hashed = false;

    private RowKey(String unique) {
        this.unique = unique;
    }

    private RowKey(RowKey rowKey) {
        this.unique = rowKey.unique;
        this.time = rowKey.time;
        this.reversed = rowKey.reversed;
        this.hashed = rowKey.hashed;
        this.args = rowKey.args;
    }

    public RowKey setTime(Long time) {
        this.time = time;
        return this;
    }

    public RowKey setArgs(String... args) {
        return setArgs(Arrays.asList(args));
    }

    public RowKey setArgs(List<String> args) {
        this.args = args;
        return this;
    }

    public RowKey reversed() {
        this.reversed = true;
        return this;
    }

    public RowKey hashed() {
        this.hashed = true;
        return this;
    }

    public byte[] build() {
        String unique = this.unique;
        if (reversed) {
            unique = new StringBuffer(unique).reverse().toString();
        }
        byte[] rowKey = Bytes.toBytes(unique);
        if (hashed) {
            rowKey = Bytes.toBytes(MD5Hash.getMD5AsHex(rowKey));
        }
        if (time != null) {
            rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
            rowKey = Bytes.add(rowKey, Bytes.toBytes(Long.MAX_VALUE - time));
        }
        if (args != null) {
            for (String arg : args) {
                if (arg == null || arg.isEmpty()) continue;
                rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
                rowKey = Bytes.add(rowKey, Bytes.toBytes(arg));
                if (arg.equals(args.get(args.size() - 1))) {
                    rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
                }
            }
        }
        return rowKey;
    }

    public static byte[] buildKey(byte[] rowKey, Long time, String... args) {
        if (time != null) {
            rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
            rowKey = Bytes.add(rowKey, Bytes.toBytes(Long.MAX_VALUE - time));
        }
        if (args != null) {
            for (String arg : args) {
                if (arg == null || arg.isEmpty()) continue;
                rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
                rowKey = Bytes.add(rowKey, Bytes.toBytes(arg));
                if (arg.equals(args[args.length - 1])) {
                    rowKey = Bytes.add(rowKey, Bytes.toBytes("_"));
                }
            }
        }
        return rowKey;
    }

    public static RowKey builder(String unique) {
        return new RowKey(unique);
    }

    public static RowKey copy(RowKey rowKey) {
        return new RowKey(rowKey);
    }

    public static byte[] valueOf(String rowKey) {
        return HBaseUtils.decode(rowKey);
    }

    public boolean isHashed() {
        return hashed;
    }

    public boolean isReversed() {
        return reversed;
    }

    public List<String> getArgs() {
        return args;
    }

    public String getUnique() {
        return unique;
    }

    public Long getTime() {
        return time;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        RowKey rowKey = (RowKey) o;
        return reversed == rowKey.reversed &&
                hashed == rowKey.hashed &&
                unique.equals(rowKey.unique) &&
                time.equals(rowKey.time) &&
                args.equals(rowKey.args);
    }

    @Override
    public int hashCode() {
        return Objects.hash(unique, time, args, reversed, hashed);
    }

    @Override
    public String toString() {
        return "RowKey{" +
                "unique='" + unique + '\'' +
                ", time=" + time +
                ", args=" + args +
                ", reversed=" + reversed +
                ", hashed=" + hashed +
                '}';
    }
}
